package org.yuanzheng.checkpoint

import org.apache.flink.runtime.state.filesystem.FsStateBackend
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.api.environment.CheckpointConfig.ExternalizedCheckpointCleanup
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}

/**
 * 使用wordcount
 */
object TestCheckPointByHDFS {
  def main(args: Array[String]): Unit = {
    def main(args: Array[String]): Unit = {
      val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
      streamEnv.setParallelism(1)
      //开启checkpoint
      streamEnv.enableCheckpointing(5000) //每隔5秒开启一次checkpoint
      streamEnv.setStateBackend(new FsStateBackend("hdfs://192.168.1.101:9000/hdfssink/")) //存放检查点数据
      //设置checkpoint参数
      streamEnv.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
      streamEnv.getCheckpointConfig.setCheckpointTimeout(5000)
      streamEnv.getCheckpointConfig.setMaxConcurrentCheckpoints(1)
      streamEnv.getCheckpointConfig.enableExternalizedCheckpoints(ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION)

      val stream: DataStream[String] = streamEnv.socketTextStream("192.168.1.8", 8888)
      import org.apache.flink.streaming.api.scala._
      stream.flatMap(_.split(" ")).map((_, 1)).keyBy(0).sum(1).print()
      streamEnv.execute("wordCount")
    }
  }
}
